package com.imooc.producer.d_trans;

import com.utils.Const;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Random;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2024/4/10-2024
 * @Description: 发送事务消息
 */
@SuppressWarnings("all")
public class TransactionProducer {
    public static void main(String[] args) throws Exception {
        //定义生产者
        TransactionMQProducer producer = new TransactionMQProducer("trans-group");
        producer.setNamesrvAddr(Const.NAME_SERVER_SINGLE);

        //定义一个 线程池
        producer.setExecutorService(Executors.newFixedThreadPool(3, new ThreadFactory() {
            private final AtomicInteger threadNum = new AtomicInteger(1);
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setName("my-thread-" + threadNum.getAndIncrement());
                return t;
            }
        }));

        //定义一个事务监听器
        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                System.err.println(Thread.currentThread().getName());
                //本地事务逻辑
                Random random = new Random();
                int num = random.nextInt(100);
                //当为7的倍数的时候，假设本地事务成功，所以提交 rocketmq 事务
                if (num % 7 == 0) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                //当为3的倍数的时候，假设本地事务失败，所以回滚 rocketmq 事务
                else if (num % 3 == 0) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                //其他情况，rocketmq 事务为 UNKNOW。 等待mq 的回查。
                else {
                    return LocalTransactionState.UNKNOW;
                }
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                //通过打印，可以看到回查的时候，会使用我们定义的线程池中的线程进行回查
                System.err.println(Thread.currentThread().getName());
                //回查 逻辑
                Random random = new Random();
                int num = random.nextInt(100);
                //当为7的倍数的时候，假设回查结果显示本地事务已经提交了，所以提交 rocketmq 事务
                if (num % 7 == 0) {
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                //当为3的倍数的时候，假设回查本地事务回滚了，所以回滚 rocketmq 事务
                else if (num % 3 == 0) {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
                //其他情况，
                else {
                    return LocalTransactionState.UNKNOW;
                }
            }
        });

        //启动生产者
        producer.start();

        //定义消息
        Message message = new Message(
                "trans_topic_msg",
                "tags_trans",
                "key_trans",
                ("trans 事务消息1").getBytes());

        //发送消息
        producer.sendMessageInTransaction(message,null);


    }
};